package netty.udp.demo3;

import cn.hutool.core.util.CharsetUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.DatagramPacket;
import io.netty.channel.socket.nio.NioDatagramChannel;
import io.netty.handler.codec.MessageToMessageEncoder;
import lombok.extern.slf4j.Slf4j;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * @program JavaBase
 * @description: 日志广播 - 客户端等待、服务端等待
 *  服务端广播 - 客户端接收 - 客户端返回 - 服务端接收
 * @author: zhanglu
 * @create: 2020-07-01 15:04:00
 */
@Slf4j
public class LogEventBroadcaster {

    public final static String HOST = "127.0.0.1";
    public final static Integer SERVER_PORT = 8000;
    public final static Integer CLIENT_PORT_1 = 8001;
    public final static Integer CLIENT_PORT_2 = 8002;
    public final static Integer CLIENT_PORT_3 = 8003;
    public final static List<Integer> clientList = Arrays.asList(CLIENT_PORT_1, CLIENT_PORT_2, CLIENT_PORT_3);

    private final Bootstrap bootstrap;
    private final File file;
    private final EventLoopGroup group;

    public LogEventBroadcaster(List<InetSocketAddress> addressList, File file) {
        this.group = new NioEventLoopGroup();
        this.bootstrap = new Bootstrap();
        this.bootstrap.group(this.group)
                .channel(NioDatagramChannel.class)
                .option(ChannelOption.SO_BROADCAST, true)
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                        ch.pipeline().addLast(new LogEventEncoder(addressList));
                        ch.pipeline().addLast(new LogEventHandle());
                    }
                });
        this.file = file;
    }

    /**
     * 启动
     */
    public void start() throws Exception {
        Channel channel = this.bootstrap.bind(SERVER_PORT).sync().channel();
        log.info("LogEventBroadcaster start");
        //读取日志文件，往管道发送，至 客户端
        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(new FileInputStream(file), CharsetUtil.UTF_8));
        String line = null;
        while ((line = bufferedReader.readLine()) != null){
            channel.writeAndFlush(line);
//            TimeUnit.SECONDS.sleep(1);
        }
        bufferedReader.close();
        log.info("LogEventBroadcaster end");

        channel.closeFuture().sync();
        System.out.println(1);
    }

    /**
     * 正常关机
     */
    public void stop(){
        group.shutdownGracefully();
    }

    /**
     * 消息发送客户端前，编码
     */
    private class LogEventEncoder extends MessageToMessageEncoder<String> {

        private final List<InetSocketAddress> addressList;

        private LogEventEncoder(List<InetSocketAddress> addressList) {
            this.addressList = addressList;
        }

        @Override
        protected void encode(ChannelHandlerContext ctx, String message, List<Object> out) throws Exception {
            for (InetSocketAddress inetSocketAddress : addressList) {
                //消息编码
                byte[] bytes = message.getBytes(CharsetUtil.UTF_8);
                out.add(new DatagramPacket(Unpooled.copiedBuffer(bytes), inetSocketAddress));
            }
        }

    }

    private class LogEventHandle extends SimpleChannelInboundHandler<DatagramPacket> {

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket datagramPacket) throws Exception {
            ByteBuf content = datagramPacket.copy().content();
            String msg = content.toString(CharsetUtil.charset(CharsetUtil.UTF_8));
            log.info(msg);
        }
    }

    public static void main(String[] args) {
        //消息
        String filePath = "/Users/zhanglu/Desktop/luImg/catalina.out";
        //远程广播地址
        List<InetSocketAddress> inetSocketAddressList = new ArrayList<>();
        for (Integer port : clientList) {
            InetSocketAddress inetSocketAddress = new InetSocketAddress(HOST, port);
            inetSocketAddressList.add(inetSocketAddress);
        }

        LogEventBroadcaster logEventBroadcaster = new LogEventBroadcaster(inetSocketAddressList, new File(filePath));
        try {
            logEventBroadcaster.start();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            logEventBroadcaster.stop();
        }
    }

}
